package replication

import (
	"context"
	"fmt"
	"net/http"
	"sync"
	"time"

	"github.com/louloulin/dataflare/pkg/processing/data"
)

// ReplicationMode 表示复制模式
type ReplicationMode int

const (
	// ReplicationModeSync 表示同步复制
	ReplicationModeSync ReplicationMode = iota
	// ReplicationModeAsync 表示异步复制
	ReplicationModeAsync
	// ReplicationModeSemiSync 表示半同步复制
	ReplicationModeSemiSync
)

// String 返回复制模式的字符串表示
func (m ReplicationMode) String() string {
	switch m {
	case ReplicationModeSync:
		return "sync"
	case ReplicationModeAsync:
		return "async"
	case ReplicationModeSemiSync:
		return "semi-sync"
	default:
		return "unknown"
	}
}

// ReplicationState 表示复制状态
type ReplicationState int

const (
	// ReplicationStateInitializing 表示初始化状态
	ReplicationStateInitializing ReplicationState = iota
	// ReplicationStateSyncing 表示同步状态
	ReplicationStateSyncing
	// ReplicationStateActive 表示活动状态
	ReplicationStateActive
	// ReplicationStatePaused 表示暂停状态
	ReplicationStatePaused
	// ReplicationStateFailed 表示失败状态
	ReplicationStateFailed
)

// String 返回复制状态的字符串表示
func (s ReplicationState) String() string {
	switch s {
	case ReplicationStateInitializing:
		return "initializing"
	case ReplicationStateSyncing:
		return "syncing"
	case ReplicationStateActive:
		return "active"
	case ReplicationStatePaused:
		return "paused"
	case ReplicationStateFailed:
		return "failed"
	default:
		return "unknown"
	}
}

// ReplicationRole 表示复制角色
type ReplicationRole int

const (
	// ReplicationRolePrimary 表示主要角色
	ReplicationRolePrimary ReplicationRole = iota
	// ReplicationRoleReplica 表示副本角色
	ReplicationRoleReplica
	// ReplicationRoleObserver 表示观察者角色
	ReplicationRoleObserver
)

// String 返回复制角色的字符串表示
func (r ReplicationRole) String() string {
	switch r {
	case ReplicationRolePrimary:
		return "primary"
	case ReplicationRoleReplica:
		return "replica"
	case ReplicationRoleObserver:
		return "observer"
	default:
		return "unknown"
	}
}

// ReplicationConfig 表示复制配置
type ReplicationConfig struct {
	// Mode 是复制模式
	Mode ReplicationMode
	// Role 是复制角色
	Role ReplicationRole
	// GroupID 是复制组ID
	GroupID string
	// NodeID 是节点ID
	NodeID string
	// Partners 是复制伙伴列表
	Partners []string
	// SyncInterval 是同步间隔
	SyncInterval time.Duration
	// MaxBatchSize 是最大批处理大小
	MaxBatchSize int
	// AckTimeout 是确认超时时间
	AckTimeout time.Duration
	// RetryCount 是重试次数
	RetryCount int
	// RetryInterval 是重试间隔
	RetryInterval time.Duration
}

// DefaultReplicationConfig 返回默认复制配置
func DefaultReplicationConfig() *ReplicationConfig {
	return &ReplicationConfig{
		Mode:          ReplicationModeAsync,
		Role:          ReplicationRolePrimary,
		SyncInterval:  time.Second,
		MaxBatchSize:  100,
		AckTimeout:    time.Second * 5,
		RetryCount:    3,
		RetryInterval: time.Second,
	}
}

// ReplicationStats 表示复制统计信息
type ReplicationStats struct {
	// ReplicatedRecords 是已复制记录数
	ReplicatedRecords uint64
	// FailedRecords 是失败记录数
	FailedRecords uint64
	// LastReplicationTime 是最后复制时间
	LastReplicationTime time.Time
	// ReplicationLag 是复制延迟
	ReplicationLag time.Duration
	// SuccessRate 是成功率
	SuccessRate float64
}

// ReplicationPartner 表示复制伙伴
type ReplicationPartner struct {
	// NodeID 是节点ID
	NodeID string
	// Address 是地址
	Address string
	// State 是状态
	State ReplicationState
	// LastSyncTime 是最后同步时间
	LastSyncTime time.Time
	// SyncOffset 是同步偏移量
	SyncOffset uint64
	// Stats 是统计信息
	Stats ReplicationStats
}

// ReplicationTransport 表示复制传输接口
type ReplicationTransport interface {
	// Connect 连接到伙伴
	Connect(ctx context.Context, partnerID string, address string) error
	// Disconnect 断开与伙伴的连接
	Disconnect(ctx context.Context, partnerID string) error
	// SendBatch 发送批次
	SendBatch(ctx context.Context, partnerID string, batch *data.DataBatch) error
	// ReceiveBatch 接收批次
	ReceiveBatch(ctx context.Context) (*data.DataBatch, string, error)
	// SendAck 发送确认
	SendAck(ctx context.Context, partnerID string, offset uint64) error
	// ReceiveAck 接收确认
	ReceiveAck(ctx context.Context) (string, uint64, error)
}

// Replicator 表示复制器
type Replicator struct {
	// config 是复制配置
	config *ReplicationConfig
	// transport 是复制传输
	transport ReplicationTransport
	// partners 是复制伙伴映射表
	partners map[string]*ReplicationPartner
	// state 是复制状态
	state ReplicationState
	// offset 是当前偏移量
	offset uint64
	// queue 是复制队列
	queue []*data.DataRecord
	// mu 是互斥锁
	mu sync.RWMutex
	// ctx 是上下文
	ctx context.Context
	// cancel 是取消函数
	cancel context.CancelFunc
	// wg 是等待组
	wg sync.WaitGroup
}

// NewReplicator 创建一个新的复制器
func NewReplicator(config *ReplicationConfig, transport ReplicationTransport) *Replicator {
	if config == nil {
		config = DefaultReplicationConfig()
	}

	ctx, cancel := context.WithCancel(context.Background())

	return &Replicator{
		config:    config,
		transport: transport,
		partners:  make(map[string]*ReplicationPartner),
		state:     ReplicationStateInitializing,
		offset:    0,
		queue:     make([]*data.DataRecord, 0),
		ctx:       ctx,
		cancel:    cancel,
	}
}

// Start 启动复制器
func (r *Replicator) Start() error {
	r.mu.Lock()
	defer r.mu.Unlock()

	if r.state != ReplicationStateInitializing && r.state != ReplicationStatePaused && r.state != ReplicationStateFailed {
		return fmt.Errorf("replicator is already running")
	}

	// 连接到所有伙伴
	for _, partnerID := range r.config.Partners {
		partner, ok := r.partners[partnerID]
		if !ok {
			// 创建新伙伴
			partner = &ReplicationPartner{
				NodeID:       partnerID,
				State:        ReplicationStateInitializing,
				LastSyncTime: time.Time{},
				SyncOffset:   0,
				Stats:        ReplicationStats{},
			}
			r.partners[partnerID] = partner
		}

		// 连接到伙伴
		err := r.transport.Connect(r.ctx, partnerID, partner.Address)
		if err != nil {
			return fmt.Errorf("failed to connect to partner %s: %w", partnerID, err)
		}

		// 更新伙伴状态
		partner.State = ReplicationStateSyncing
	}

	// 更新状态
	r.state = ReplicationStateActive

	// 启动复制循环
	if r.config.Role == ReplicationRolePrimary {
		r.wg.Add(1)
		go r.replicationLoop()
	}

	// 启动接收循环
	r.wg.Add(1)
	go r.receiveLoop()

	return nil
}

// Stop 停止复制器
func (r *Replicator) Stop() error {
	r.mu.Lock()
	defer r.mu.Unlock()

	if r.state == ReplicationStateInitializing || r.state == ReplicationStatePaused || r.state == ReplicationStateFailed {
		return fmt.Errorf("replicator is not running")
	}

	// 取消上下文
	r.cancel()

	// 等待所有协程结束
	r.wg.Wait()

	// 断开与所有伙伴的连接
	for partnerID, partner := range r.partners {
		err := r.transport.Disconnect(context.Background(), partnerID)
		if err != nil {
			return fmt.Errorf("failed to disconnect from partner %s: %w", partnerID, err)
		}
		partner.State = ReplicationStatePaused
	}

	// 更新状态
	r.state = ReplicationStatePaused

	return nil
}

// AddPartner 添加复制伙伴
func (r *Replicator) AddPartner(partnerID, address string) error {
	r.mu.Lock()
	defer r.mu.Unlock()

	// 检查伙伴是否已存在
	if _, ok := r.partners[partnerID]; ok {
		return fmt.Errorf("partner %s already exists", partnerID)
	}

	// 创建新伙伴
	partner := &ReplicationPartner{
		NodeID:       partnerID,
		Address:      address,
		State:        ReplicationStateInitializing,
		LastSyncTime: time.Time{},
		SyncOffset:   0,
		Stats:        ReplicationStats{},
	}
	r.partners[partnerID] = partner

	// 如果复制器已启动，连接到伙伴
	if r.state == ReplicationStateActive {
		err := r.transport.Connect(r.ctx, partnerID, address)
		if err != nil {
			delete(r.partners, partnerID)
			return fmt.Errorf("failed to connect to partner %s: %w", partnerID, err)
		}
		partner.State = ReplicationStateSyncing
	}

	return nil
}

// RemovePartner 移除复制伙伴
func (r *Replicator) RemovePartner(partnerID string) error {
	r.mu.Lock()
	defer r.mu.Unlock()

	// 检查伙伴是否存在
	_, ok := r.partners[partnerID]
	if !ok {
		return fmt.Errorf("partner %s not found", partnerID)
	}

	// 如果复制器已启动，断开与伙伴的连接
	if r.state == ReplicationStateActive {
		err := r.transport.Disconnect(r.ctx, partnerID)
		if err != nil {
			return fmt.Errorf("failed to disconnect from partner %s: %w", partnerID, err)
		}
	}

	// 移除伙伴
	delete(r.partners, partnerID)

	// 更新配置
	for i, id := range r.config.Partners {
		if id == partnerID {
			r.config.Partners = append(r.config.Partners[:i], r.config.Partners[i+1:]...)
			break
		}
	}

	return nil
}

// GetPartner 获取复制伙伴
func (r *Replicator) GetPartner(partnerID string) (*ReplicationPartner, bool) {
	r.mu.RLock()
	defer r.mu.RUnlock()

	partner, ok := r.partners[partnerID]
	return partner, ok
}

// GetAllPartners 获取所有复制伙伴
func (r *Replicator) GetAllPartners() []*ReplicationPartner {
	r.mu.RLock()
	defer r.mu.RUnlock()

	partners := make([]*ReplicationPartner, 0, len(r.partners))
	for _, partner := range r.partners {
		partners = append(partners, partner)
	}
	return partners
}

// GetState 获取复制状态
func (r *Replicator) GetState() ReplicationState {
	r.mu.RLock()
	defer r.mu.RUnlock()
	return r.state
}

// GetOffset 获取当前偏移量
func (r *Replicator) GetOffset() uint64 {
	r.mu.RLock()
	defer r.mu.RUnlock()
	return r.offset
}

// GetStats 获取复制统计信息
func (r *Replicator) GetStats() map[string]ReplicationStats {
	r.mu.RLock()
	defer r.mu.RUnlock()

	stats := make(map[string]ReplicationStats)
	for partnerID, partner := range r.partners {
		stats[partnerID] = partner.Stats
	}
	return stats
}

// Replicate 复制记录
func (r *Replicator) Replicate(record *data.DataRecord) error {
	r.mu.Lock()
	defer r.mu.Unlock()

	if r.state != ReplicationStateActive {
		return fmt.Errorf("replicator is not active")
	}

	if r.config.Role != ReplicationRolePrimary {
		return fmt.Errorf("only primary can replicate data")
	}

	// 添加记录到队列
	r.queue = append(r.queue, record)

	return nil
}

// ReplicateBatch 复制批次
func (r *Replicator) ReplicateBatch(batch *data.DataBatch) error {
	r.mu.Lock()
	defer r.mu.Unlock()

	if r.state != ReplicationStateActive {
		return fmt.Errorf("replicator is not active")
	}

	if r.config.Role != ReplicationRolePrimary {
		return fmt.Errorf("only primary can replicate data")
	}

	// 添加记录到队列
	for _, record := range batch.Records {
		r.queue = append(r.queue, record)
	}

	return nil
}

// replicationLoop 复制循环
func (r *Replicator) replicationLoop() {
	defer r.wg.Done()

	ticker := time.NewTicker(r.config.SyncInterval)
	defer ticker.Stop()

	for {
		select {
		case <-r.ctx.Done():
			return
		case <-ticker.C:
			r.syncBatch()
		}
	}
}

// syncBatch 同步批次
func (r *Replicator) syncBatch() {
	r.mu.Lock()
	// 检查队列是否为空
	if len(r.queue) == 0 {
		r.mu.Unlock()
		return
	}

	// 获取批次大小
	batchSize := r.config.MaxBatchSize
	if batchSize > len(r.queue) {
		batchSize = len(r.queue)
	}

	// 创建批次
	batch := data.NewDataBatch()
	for i := 0; i < batchSize; i++ {
		batch.AddRecord(r.queue[i])
	}

	// 获取伙伴列表
	partners := make([]string, 0, len(r.partners))
	for partnerID := range r.partners {
		partners = append(partners, partnerID)
	}
	r.mu.Unlock()

	// 同步到所有伙伴
	for _, partnerID := range partners {
		r.syncToPartner(partnerID, batch)
	}

	// 更新队列
	r.mu.Lock()
	r.queue = r.queue[batchSize:]
	r.offset += uint64(batchSize)
	r.mu.Unlock()
}

// syncToPartner 同步到伙伴
func (r *Replicator) syncToPartner(partnerID string, batch *data.DataBatch) {
	r.mu.RLock()
	partner, ok := r.partners[partnerID]
	if !ok {
		r.mu.RUnlock()
		return
	}
	r.mu.RUnlock()

	// 发送批次
	err := r.transport.SendBatch(r.ctx, partnerID, batch)
	if err != nil {
		// 更新统计信息
		r.mu.Lock()
		partner.Stats.FailedRecords += uint64(batch.Count())
		r.mu.Unlock()
		return
	}

	// 如果是同步模式，等待确认
	if r.config.Mode == ReplicationModeSync {
		// 等待确认
		ackCtx, cancel := context.WithTimeout(r.ctx, r.config.AckTimeout)
		defer cancel()

		for i := 0; i < r.config.RetryCount; i++ {
			// 接收确认
			ackPartnerID, offset, err := r.transport.ReceiveAck(ackCtx)
			if err != nil {
				// 重试
				time.Sleep(r.config.RetryInterval)
				continue
			}

			// 检查确认
			if ackPartnerID == partnerID {
				// 更新伙伴状态
				r.mu.Lock()
				partner.LastSyncTime = time.Now()
				partner.SyncOffset = offset
				partner.Stats.ReplicatedRecords += uint64(batch.Count())
				partner.Stats.LastReplicationTime = time.Now()
				r.mu.Unlock()
				return
			}
		}

		// 更新统计信息
		r.mu.Lock()
		partner.Stats.FailedRecords += uint64(batch.Count())
		r.mu.Unlock()
	} else {
		// 异步模式，直接更新统计信息
		r.mu.Lock()
		partner.Stats.ReplicatedRecords += uint64(batch.Count())
		partner.Stats.LastReplicationTime = time.Now()
		r.mu.Unlock()
	}
}

// receiveLoop 接收循环
func (r *Replicator) receiveLoop() {
	defer r.wg.Done()

	for {
		select {
		case <-r.ctx.Done():
			return
		default:
			// 接收批次
			batch, partnerID, err := r.transport.ReceiveBatch(r.ctx)
			if err != nil {
				// 检查是否是上下文取消错误
				if r.ctx.Err() != nil {
					return
				}
				// 记录错误并继续
				fmt.Printf("Failed to receive batch: %v\n", err)
				time.Sleep(r.config.RetryInterval)
				continue
			}

			// 处理批次
			r.processBatch(batch, partnerID)
		}
	}
}

// processBatch 处理批次
func (r *Replicator) processBatch(batch *data.DataBatch, partnerID string) {
	r.mu.Lock()
	defer r.mu.Unlock()

	// 检查角色
	if r.config.Role != ReplicationRoleReplica {
		// 主节点收到批次，可能是错误的
		return
	}

	// 更新偏移量
	r.offset += uint64(batch.Count())

	// 发送确认
	if r.config.Mode == ReplicationModeSync {
		err := r.transport.SendAck(r.ctx, partnerID, r.offset)
		if err != nil {
			fmt.Printf("Failed to send ack: %v\n", err)
		}
	}
}

// HTTPReplicationTransport 表示HTTP复制传输
type HTTPReplicationTransport struct {
	// baseURL 是基础URL
	baseURL string
	// client 是HTTP客户端
	client *http.Client
	// receiveCh 是接收通道
	receiveCh chan *ReplicationMessage
	// ackCh 是确认通道
	ackCh chan *ReplicationAck
}

// ReplicationMessage 表示复制消息
type ReplicationMessage struct {
	// PartnerID 是伙伴ID
	PartnerID string
	// Batch 是数据批次
	Batch *data.DataBatch
}

// ReplicationAck 表示复制确认
type ReplicationAck struct {
	// PartnerID 是伙伴ID
	PartnerID string
	// Offset 是偏移量
	Offset uint64
}

// NewHTTPReplicationTransport 创建一个新的HTTP复制传输
func NewHTTPReplicationTransport(baseURL string) *HTTPReplicationTransport {
	return &HTTPReplicationTransport{
		baseURL:   baseURL,
		client:    &http.Client{},
		receiveCh: make(chan *ReplicationMessage, 100),
		ackCh:     make(chan *ReplicationAck, 100),
	}
}

// Connect 连接到伙伴
func (t *HTTPReplicationTransport) Connect(ctx context.Context, partnerID string, address string) error {
	// 在实际实现中，这里应该建立HTTP连接
	return nil
}

// Disconnect 断开与伙伴的连接
func (t *HTTPReplicationTransport) Disconnect(ctx context.Context, partnerID string) error {
	// 在实际实现中，这里应该关闭HTTP连接
	return nil
}

// SendBatch 发送批次
func (t *HTTPReplicationTransport) SendBatch(ctx context.Context, partnerID string, batch *data.DataBatch) error {
	// 在实际实现中，这里应该通过HTTP发送批次
	// 为了测试，我们直接发送到接收通道
	select {
	case t.receiveCh <- &ReplicationMessage{PartnerID: partnerID, Batch: batch}:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

// ReceiveBatch 接收批次
func (t *HTTPReplicationTransport) ReceiveBatch(ctx context.Context) (*data.DataBatch, string, error) {
	// 在实际实现中，这里应该通过HTTP接收批次
	// 为了测试，我们直接从接收通道接收
	select {
	case msg := <-t.receiveCh:
		return msg.Batch, msg.PartnerID, nil
	case <-ctx.Done():
		return nil, "", ctx.Err()
	}
}

// SendAck 发送确认
func (t *HTTPReplicationTransport) SendAck(ctx context.Context, partnerID string, offset uint64) error {
	// 在实际实现中，这里应该通过HTTP发送确认
	// 为了测试，我们直接发送到确认通道
	select {
	case t.ackCh <- &ReplicationAck{PartnerID: partnerID, Offset: offset}:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

// ReceiveAck 接收确认
func (t *HTTPReplicationTransport) ReceiveAck(ctx context.Context) (string, uint64, error) {
	// 在实际实现中，这里应该通过HTTP接收确认
	// 为了测试，我们直接从确认通道接收
	select {
	case ack := <-t.ackCh:
		return ack.PartnerID, ack.Offset, nil
	case <-ctx.Done():
		return "", 0, ctx.Err()
	}
}
